package com.arxan;

import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.Reader;
import java.net.URISyntaxException;
import java.util.Properties;
import org.apache.log4j.Logger;
import org.apache.log4j.xml.DOMConfigurator;

/**
 * Hello world!
 *
 */
public class App 
{
	public HDFSFetch hdfsclient;
	private Properties properties = new Properties();
	public String hdfs_url;
	public String loc_prefix ;
	public String hdfs_user ;
	public String post_url;
	private static Logger log;
	public CalcSpeed calcSpeed = new CalcSpeed();
	public DBSaveStep dbsave = null;
	//private static Logger logger = Logger.getLogger(App.class);
	
    public App() throws IOException {
		super();
		FileInputStream pfile = new FileInputStream("config.prop");
    	properties.load(pfile); 
    	pfile.close();
    	hdfs_url = properties.getProperty("hdfs_url");
    	loc_prefix = properties.getProperty("hdfs_loc_prefix");
    	hdfs_user = properties.getProperty("hdfs_user");
    	post_url = properties.getProperty("http_post_url");
    	
    	log.info("hdfs url is:" + hdfs_url + " loc_prefix:" + loc_prefix + "hdfs_user:" + hdfs_user);
	}
    /**
     * 从hdfs获取数据，发送到datapipe
     *
     * @param 
     * @return
     */
	private void hdfsToDatapipe() throws URISyntaxException, IOException {
		
		//CalcSpeed Thread start
        Thread calcTh = new Thread(calcSpeed);
        calcTh.start();
        
		hdfsclient = new HDFSFetch(hdfs_url,hdfs_user);
    	String[] dataTypes = hdfsclient.getFileList(loc_prefix);
    	for(String datatype:dataTypes){
    		String curdir = loc_prefix+"/" + datatype;
    		log.info("Process P2P:" + curdir);
    		String [] p2pcompanys = hdfsclient.getFileList(curdir);
    		for( String p2pcompany : p2pcompanys){
    			String subdir = curdir +"/" +  p2pcompany;
    			log.info("Process P2P And DataType :"+ subdir);
    			String[] procdates = hdfsclient.getFileList(subdir);
    			for(String procdate : procdates){
    				String p2p_type_date = subdir + "/" + procdate;
    				String[] files = hdfsclient.getFileList(p2p_type_date);
    				for(String targetfile : files){
    					String proctargetfile = p2p_type_date + "/" + targetfile;
    					int  oldstep = dbsave.getStep(proctargetfile);
    					if(oldstep == -1)
    						oldstep = 0;
    					log.debug("Get file " + proctargetfile + "old step " + oldstep);
    					ProcessTargetFile(p2pcompany,datatype,proctargetfile,oldstep);
    				}
    			}
    		}
    		
    	}
    	
    	calcSpeed.stopThread();
    	
    	try {
    		calcTh.join();
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		
	}
	
	/**
     * 处理单个文件
     *
     * @param p2pcompany
     *            公司编号
     * @param datatype
     *            数据类型，包括transcationdata,transfer,scatteredinvest,updatestatus,financing
     * @param proctargetfile
     * 			  文件名
     * @param oldstep
     * 			  在数据库中保存的，已经处理过的行数
     * @return 
     */
	private void ProcessTargetFile(String p2pcompany,String datatype,String proctargetfile, int oldstep) {
		log.debug("http post url is:" + post_url);
		try {
			log.debug("file " + proctargetfile + " old step: " + oldstep);
			int proccount = hdfsclient.ProcessTarget(post_url,p2pcompany,datatype,proctargetfile,calcSpeed,oldstep,dbsave);
			log.debug("processed " + proccount + " items");
		} catch (IllegalArgumentException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		
	}

	public static void main( String[] args ) throws IOException, URISyntaxException
    {
    	DOMConfigurator.configure("./log4j.xml");
    	log=Logger.getLogger("railyFile");
    	
    	App app = new App();
    	app.dbsave = new DBSaveStep("dbstep.db");
    	
    	app.hdfsToDatapipe();
    	
    	//app.sendExample();
    	
    	System.out.println("End of Process");
    }

	/**
     * 用于测试一个json发送的代码
     */
	@SuppressWarnings("unused")
	private void sendExample() throws IOException {
		String trueposturl = post_url+"financing";
		String sampleFileName = "./examples/sample.json";
		log=Logger.getLogger("railyFile");
		log.info("filename:"+sampleFileName);
		log.info("post url:"+trueposturl);
		FileInputStream in = new FileInputStream(sampleFileName);
		char[] tempchars = new char[in.available()];
        
        Reader reader = new InputStreamReader(in);
        reader.read(tempchars);
        reader.close();
        in.close();
		String postdata = new String(tempchars);
		String response = DataPipeClient.sendPost(trueposturl, postdata);
		log.info("Response:" + response );
	}
	
}
